package com.cyzy.sender;

import com.alibaba.fastjson.JSON;
import com.cyzy.dto.Message;
import com.cyzy.dto.MessageType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;

// 聊天消息生产者，支持同步和异步发送
@Component // 声明为Spring组件，方便注入
public class ChatProducer {
    @Value("${rocketmq.name-server}")
    private String namesrvAddr;
    
    @Value("${rocketmq.producer.group}")
    private String producerGroup;
    
    private static final String TOPIC = "producer_topic";

    private DefaultMQProducer producer;

    @PostConstruct
    public void init() throws MQClientException {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        // 增加生产者配置，优化性能和可靠性
        producer.setRetryTimesWhenSendFailed(3); // 同步发送失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(3); // 异步发送失败重试次数
        producer.setSendMsgTimeout(3000); // 发送超时时间
        producer.start();
        System.out.println("ChatProducer初始化完成，NameServer地址: " + namesrvAddr);
    }

    // 同步发送消息（原有功能保留）
    public String sendMessage(String fromUser, Integer toUserid, String toUser, String content) throws Exception {
        Message customMessage = new Message(fromUser, toUserid, toUser, content);
        org.apache.rocketmq.common.message.Message msg = createRocketMqMessage(toUser, customMessage);

        SendResult result = producer.send(msg);
        System.out.printf("同步消息发送成功: %s%n", result.getMsgId());
        return result.getMsgId();
    }

    // 新增：异步发送消息
    public void sendMessageAsync(String fromUser,Integer toUserid ,String toUser, String content,MessageType  messageType, Integer messageId, SendCallback callback) throws Exception {
        Message customMessage = new Message(fromUser, toUserid,toUser, content);
        customMessage.setMessageid(messageId);
        customMessage.setMessageType(messageType);
        org.apache.rocketmq.common.message.Message msg = createRocketMqMessage(toUser, customMessage);
        // 异步发送，立即返回，结果通过回调处理
        producer.send(msg, callback);
        System.out.println("异步消息发送请求已提交，消息ID: " + messageId);
    }

    //创建RocketMQ消息对象
    private org.apache.rocketmq.common.message.Message createRocketMqMessage(String toUser, Message customMessage) throws Exception {
        return new org.apache.rocketmq.common.message.Message(
                TOPIC,
                toUser,
                JSON.toJSONString(customMessage).getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
    }

    //销毁生产者
    @PreDestroy
    public void shutdown() {
        if (producer != null) {
            producer.shutdown();
            System.out.println("ChatProducer已关闭");
        }
    }

}